package com.zetyun.tiger.app;

import com.ververica.cdc.connectors.oracle.OracleSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Properties;

public class OracleSourceExample {

    public static void main(String[] args) throws Exception {

        Properties prop = new Properties();
        prop.put("database.tablename.case.insensitive","false");
//        prop.put("database.url","jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS_LIST=(LOAD_BALANCE=NO)(FAILOVER=YES)(ADDRESS=(PROTOCOL=tcp)(HOST=rdx2)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=ORCL11G)))");
        SourceFunction<String> sourceFunction = OracleSource.<String>builder()
                .hostname("rt3")
                .port(1521)
                .database("orcl") // monitor XE database
                .schemaList("FLINKUSER") // monitor inventory schema
                .tableList("FLINKUSER.DEPT") // monitor products table
                .username("flinkuser")
                .password("flinkpw")
                .debeziumProperties(prop)
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(sourceFunction)
                .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

        env.execute();
    }
}
